package broker

import (
	"fmt"
	protocolImpl "gitee.com/tjccc/gomq/mq/protocol"
	"sync"

	concurrencyMap "github.com/seymourtang/concurrency-map"

	"gitee.com/tjccc/gomq/interface/protocol"
	"gitee.com/tjccc/gomq/lib/logger"
	conn "gitee.com/tjccc/gomq/mq/connection"
)

type Config struct {
	QueueSize int32
}

var config Config

type BrokerEngine struct {
	lock           sync.Mutex
	consumerGroups map[string]*ConsumerGroup
	subscriptions  map[string][]*ConsumerGroup // topicName-ConsumerGroup
	topics         concurrencyMap.ConcurrentMap
}

func NewBrokerEngine() *BrokerEngine {
	return &BrokerEngine{
		consumerGroups: map[string]*ConsumerGroup{},
		subscriptions:  map[string][]*ConsumerGroup{},
		topics:         concurrencyMap.New(),
	}
}

func (e *BrokerEngine) Exec(client *conn.Connection, msg protocol.Msg) (result protocol.Msg) {
	customMsg, ok := msg.(*protocolImpl.CustomMsg)
	if !ok {
		logger.Errorf("unsupported protocol")
	}
	return e.execCommand(client, customMsg.Code, customMsg.ExtMap, customMsg.Data)
}

func (e *BrokerEngine) execCommand(client *conn.Connection, code protocolImpl.Code, extMap map[string]any, data []byte) (result protocol.Msg) {
	if code == protocolImpl.SEND {
		return SendFunc(e, extMap, data)
	} else if code == protocolImpl.PULL {
		return PullFunc(e, extMap, data)
	}
	return
}

// 并发
func (e *BrokerEngine) DeclareTopic(topicName string) (*Topic, bool) {
	value, ok := e.topics.Get(topicName)
	topic, ok := value.(*Topic)
	if !ok {
		topic = NewTopic(topicName)
		e.topics.Set(topicName, topic)
		return topic, true
	}
	return topic, false
}

func (e *BrokerEngine) ReceiveMsg(topicName string, data []byte) (result *protocolImpl.CustomMsg) {
	value, ok := e.topics.Get(topicName)
	topic, ok := value.(*Topic)
	if !ok {
		return &protocolImpl.CustomMsg{
			Code: protocolImpl.ERR,
			Data: []byte(fmt.Sprintf("name[%s] not exist", topicName)),
		}
	}
	topic.PutMsg(data)
	return nil
}

func (e BrokerEngine) Subscribe(client conn.Connection, groupName string, topicName string) {
	// 创建消费组
	cg := e.GetConsumerGroup(groupName, true)
	cg.AddConsumer(client)
	e.consumerGroups[groupName] = cg
	// 创建 name
	topic, ok := e.DeclareTopic(topicName)
	// 订阅关联
	subscription, ok := e.subscriptions[topicName]
	if !ok {
		subscription = make([]*ConsumerGroup, 0)
	}
	subscription = append(subscription, cg)
	e.subscriptions[topicName] = subscription
	topic.AddSubscription(cg)
	return
}

func (e BrokerEngine) GetConsumerGroup(groupName string, create bool) *ConsumerGroup {
	cg, ok := e.consumerGroups[groupName]
	if !ok {
		cg = NewConsumerGroup(groupName)
		e.consumerGroups[groupName] = cg
	}
	return cg
}

func (e BrokerEngine) Lock() {
	e.lock.Lock()
}

func (e BrokerEngine) Unlock() {
	e.lock.Unlock()
}

func (e BrokerEngine) Close() {
}
